package db.nio.worker;

import db.nio.ChannelContext;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 可终止线程
 * Created by daibin on 16-8-23.
 */
public abstract class Worker extends Thread {

    /**
     * 线程池
     */
    protected ThreadPoolExecutor poolExecutor;

    /**
     * 中止标记
     */
    public volatile boolean stop = false;


    /**
     * 等待标记
     */
    public volatile boolean waite = false;


    /**
     * 监听器
     */
    public volatile Selector selector;


    /**
     * 构造函数
     */
    public Worker(ThreadPoolExecutor poolExecutor) {
        this.poolExecutor = poolExecutor;
        try {
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 内部线程
     */
    public void run() {
        int count = 0;
        while (true) {

            if (stop) {
                try {
                    selector.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    break;
                }
            }

            //循环等待
            if (waite) {
                continue;
            }

            try {
                long start = System.currentTimeMillis();
                int nKeys = selector.select(1);
                if (nKeys > 0) {
                    Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                    while (keyIterator.hasNext()) {
                        count = 0;
                        SelectionKey key = keyIterator.next();
                        keyIterator.remove();
                        work(key);
                    }
                }

                //检查死锁
                long split = System.currentTimeMillis() - start;
                if (split < 1) {
                    if (count++ > 100) {
                        count = 0;
                        resetSelector();
                        //System.out.println("reset selector");
                    }
                }
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 工作函数
     */
    public abstract void work(SelectionKey key);

    /**
     * epoll bug重新绑定监听器
     */
    public void resetSelector() {
        try {
            Selector newSelector = Selector.open();
            for (SelectionKey key : selector.keys()) {
                ChannelContext context = (ChannelContext) key.attachment();

                try {
                    if (key.channel().keyFor(newSelector) != null)
                        continue;
                    int interestOps = key.interestOps();
                    key.cancel();
                    key = key.channel().register(newSelector, interestOps, context);
                    context.getFramer().setSelectionKey(key);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            selector.close();
            selector = newSelector;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
